package com.snowsea.listener;

import com.alibaba.fastjson.JSON;
import com.snowsea.enums.OrderEnums;
import com.snowsea.pojo.Order;
import com.snowsea.service.OrderService;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @className: TransactionListenerImpl
 * @auther: Jerry
 * @date: 2021/4/19 17:40
 * @desctiption: TODO
 * @version: 1.0
 */
@RocketMQTransactionListener(txProducerGroup = "myTransactionGroup")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private ConcurrentHashMap<String, RocketMQLocalTransactionState> localTrans = new ConcurrentHashMap<>();

    private Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);

    @Autowired
    private OrderService orderService;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        System.out.println("half消息确认。。。");
        String orderStr = new String((byte[]) message.getPayload());
        Order order = JSON.parseObject(orderStr, Order.class);
        String transId = (String) message.getHeaders().get("rocketmq_" + RocketMQHeaders.TRANSACTION_ID);
        // 更新订单状态
        int count = orderService.updateOrderState(order.getId(), OrderEnums.PAIED.getCode());
        try {
            if (count > 0) {
                localTrans.put(transId, RocketMQLocalTransactionState.COMMIT);
            }
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            logger.error("发生异常，异常信息：{}", e.getMessage());
            // 发生异常，回滚事务
            orderService.updateOrderState(order.getId(), OrderEnums.CANCEL.getCode());
            localTrans.put(transId, RocketMQLocalTransactionState.ROLLBACK);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        System.out.println("事务消息回查。。。");
        String transId = (String) message.getHeaders().get("rocketmq_" + RocketMQHeaders.TRANSACTION_ID);
        System.out.println("transId：" + transId);
        return localTrans.get(transId);
    }
}
